from kafka import KafkaConsumer
import json

# Kafka配置
KAFKA_SERVER = '192.168.162.118:9092'
TOPIC_NAME = 'LAB-HTTP'
TARGET_IPS = ['192.168.5.241', '172.16.3.135']


def contains_target_ips(message_content):
    """检查消息内容是否包含目标IP地址"""
    if not message_content:
        return False, []
    content_str = str(message_content).lower()
    found_ips = [ip for ip in TARGET_IPS if ip in content_str]
    return len(found_ips) > 0, found_ips


def decode_message_content(message):
    """尝试多种方式解码消息内容"""
    if not message.value:
        return ""

    # 尝试UTF-8解码
    try:
        return message.value.decode('utf-8')
    except UnicodeDecodeError:
        pass

    # 尝试其他编码
    for encoding in ['latin1', 'gbk', 'gb2312']:
        try:
            return message.value.decode(encoding)
        except UnicodeDecodeError:
            continue

    # 如果都失败，返回原始字节的字符串表示
    return str(message.value)


def main():
    # 创建Kafka消费者
    print("正在初始化Kafka消费者...")
    try:
        consumer = KafkaConsumer(
            TOPIC_NAME,
            bootstrap_servers=[KAFKA_SERVER],
            auto_offset_reset='latest',  # 从最新消息开始
            consumer_timeout_ms=15000,  # 15秒超时
            enable_auto_commit=True,
            group_id='kafka_debug_consumer'
        )
        print(f"Kafka消费者初始化成功")
    except Exception as e:
        print(f"Kafka消费者初始化失败: {e}")
        return

    print(f"开始监听Kafka topic: {TOPIC_NAME}")
    print(f"服务器: {KAFKA_SERVER}")
    print(f"筛选条件: 包含IP地址 {TARGET_IPS}")
    print("=" * 50)

    try:
        matched_messages = []
        message_count = 0
        max_messages = 100  # 最大检查消息数

        print("开始读取消息...")
        for message in consumer:
            message_count += 1

            # 解码消息内容
            message_content = decode_message_content(message)

            # 检查是否包含目标IP
            has_target_ip, found_ips = contains_target_ips(message_content)

            # 显示每条消息的基本信息（前10条）
            if message_count <= 10:
                print(f"[{message_count}] 分区:{message.partition} 偏移量:{message.offset} 包含目标IP:{has_target_ip}")
                if message_content:
                    preview = message_content[:100].replace('\n', ' ').replace('\r', ' ')
                    print(f"    内容预览: {preview}...")

            if has_target_ip:
                matched_messages.append({
                    'partition': message.partition,
                    'offset': message.offset,
                    'found_ips': found_ips,
                    'content': message_content
                })
                print(
                    f"✓ 找到匹配消息 {len(matched_messages)}: 分区 {message.partition}, 偏移量 {message.offset}, 匹配IP: {found_ips}")

                # 当找到足够的匹配消息时停止
                if len(matched_messages) >= 3:
                    print(f"已找到3条匹配消息，停止搜索")
                    break

            # 设置上限防止无限循环
            if message_count >= max_messages:
                print(f"已检查{max_messages}条消息，停止搜索")
                break

        # 显示结果
        print("\n" + "=" * 50)
        print(f"总共检查了 {message_count} 条消息")
        print(f"找到 {len(matched_messages)} 条符合条件的消息:")

        if matched_messages:
            for i, msg in enumerate(matched_messages, 1):
                print(f"\n--- 匹配消息 {i} ---")
                print(f"分区: {msg['partition']}, 偏移量: {msg['offset']}")
                print(f"匹配的IP: {msg['found_ips']}")
                if msg['content']:
                    content_preview = msg['content'][:800]
                    print("内容预览:")
                    print(content_preview)
                    if len(msg['content']) > 800:
                        print("... (内容过长已截断)")
                else:
                    print("内容: (空)")
        else:
            print("未找到包含指定IP地址的消息")
            if message_count > 0:
                print("建议:")
                print("1. 检查目标IP地址是否正确")
                print("2. 尝试使用 'earliest' 模式从头开始读取消息")
                print("3. 确认Kafka topic中有数据产生")

    except Exception as e:
        print(f"错误: {e}")
        import traceback
        traceback.print_exc()
    finally:
        try:
            consumer.close()
        except:
            pass
        print("\n完成。")


if __name__ == "__main__":
    main()
